package com.career.syn;

import com.alibaba.fastjson.JSONObject;
import com.career.syn.constant.Constant;
import com.career.syn.util.CommonUtil;
import com.career.syn.util.DateUtil;
import com.career.syn.util.HmacShaUtil;
import com.career.syn.util.JsonMapper;
import com.career.syn.web.Resp;
import com.career.syn.web.RespCode;
import com.career.syn.web.SynResult;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.converter.StringHttpMessageConverter;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;
import sun.misc.BASE64Encoder;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

import static com.career.syn.util.CommonUtil.generateNonce;

/**
 * @author jasongu
 * @Description: 数据同步
 * @ClassName: DataSyn
 * @date 2019/6/19 14:11
 */
@SuppressWarnings("restriction")
@RestController
@RequestMapping("/syn")
@ComponentScan(basePackages = {"com.career.syn"})
public class DataSyn {

    private static Logger sync_log = LoggerFactory.getLogger("sync_log");

    private static Logger dump_log = LoggerFactory.getLogger("dump_log");

    private static Logger logger = LoggerFactory.getLogger(DataSyn.class);

    @Value("${api_key}")
    private String apiKey;

    @Value("${secret_key}")
    private String secretKey;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @PostMapping
    public void syn(@RequestBody SynConfig synConfig) {

        String requestUrl = synConfig.getRequestUrl();

        String path = synConfig.getPath();

        Integer permits = synConfig.getPermits();

        Integer nThreads = synConfig.getnThreads();

        Boolean haveUri = synConfig.getHaveUri();

        logger.info("文件路径:{}", path);
        logger.info("信号量:{}", permits);
        logger.info("初始化线程:{}", nThreads);
        List<File> files = new ArrayList<>();
        readDirectory(path, files);
        int i = 0;
        SynResult result = new SynResult();
        long startTime = System.currentTimeMillis();
        while (i < files.size()) {
//            if (files.get(i).getName().endsWith(".DS_Store")){
//                continue;
//            }
            boolean completed = readFile02(requestUrl, files.get(i), permits, nThreads, haveUri, result);
            if (!completed) {
                break;
            } else {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                i++;
            }
        }
        logger.info("----------------文件路径:{},所有文件读取并同步完成,总行数:{},成功数量:{},失败数量:{},总耗时:{}ms-------------------", path, result.getTotal(), result.getSuccessCount(), result.getErrorCount(), System.currentTimeMillis() - startTime);
    }

    public List<File> readDirectory(String filePath, List<File> files) {
        File file = new File(filePath);
        if (file.isDirectory()) {
            String[] list = file.list();
            for (String s : list) {
                File readFile = new File(filePath + File.separator + s);
                if (!readFile.isDirectory()) {
                    files.add(readFile);
                } else {
                    readDirectory(readFile.getPath(), files);
                }
            }
        } else {
            files.add(file);
        }
        return files;
    }

    /**
     * 读取一个文本 一行一行读取
     *
     * @param file
     * @param permits
     */
    public boolean readFile02(String requestUrl, File file, int permits, int nThreads, Boolean haveUri, SynResult result) {
        // 使用一个字符串集合来存储文本中的路径 ，也可用String []数组
        logger.info("----------------开始读取并同步文件数据:{}-------------------", file.getPath());
        FileInputStream fis = null;
        InputStreamReader isr = null;
        BufferedReader br = null;
        ExecutorService pool = Executors.newFixedThreadPool(nThreads);
        final Semaphore semaphore = new Semaphore(permits, true);
        String jsonStr = "";
        String username = "";
//		String fileName = "/Users/xiaoweigu/Desktop/opt/api_data/" + file;
        try {
            fis = new FileInputStream(file);
            // 防止路径乱码 如果utf-8 乱码 改GBK eclipse里创建的txt 用UTF-8，在电脑上自己创建的txt 用GBK
            isr = new InputStreamReader(fis, "UTF-8");
            br = new BufferedReader(isr);
            String line = "";
            LineNumberReader reader = new LineNumberReader(br);
            while (!ObjectUtil.isEmpty(line = reader.readLine())) {
                result.getTotal().getAndIncrement();
                logger.info("当前行数----------" + reader.getLineNumber());
//				if (reader.getLineNumber() > 265000){
                String uri = null;
                if (haveUri) {
                    uri = line.substring(0, line.indexOf("----"));
                    jsonStr = line.substring(line.indexOf("----") + 4, line.lastIndexOf("----"));
                } else {
                    jsonStr = line.substring(0, line.lastIndexOf("----") + 4);
                }
                username = line.substring(line.lastIndexOf("----") + 4);

                ApiThread thead = new ApiThread(requestUrl, semaphore, jsonStr, username, uri, result);
                pool.execute(thead);
            }
            pool.shutdown();
            while (true) {
                if (pool.isTerminated()) {
                    logger.info("----------------读取并同步文件数据结束:{}-------------------", file.getPath());
                    return true;
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            e.printStackTrace();
        } finally {
            try {
                br.close();
                isr.close();
                fis.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        logger.error("线程池异常,读取文件:{},请求路径:{},请求参数:{},用户名:{}", file.getPath(), requestUrl, jsonStr, username);
        return false;
    }

    public class ApiThread extends Thread {
        private Semaphore semaphore;
        private String jsonStr;
        private String username;
        private String uri;
        private String requestUrl;
        private SynResult result;

        public ApiThread(String requestUrl, Semaphore semaphore, String jsonStr, String username, String uri, SynResult result) {
            this.requestUrl = requestUrl;
            this.semaphore = semaphore;
            this.jsonStr = jsonStr;
            this.username = username;
            this.uri = uri;
            this.result = result;
        }

        @Override
        public void run() {
            try {
                semaphore.acquire();// 获取信号灯许可
                Date now = new Date();
                logger.info(Thread.currentThread() + ":" + DateUtil.format(now));
                runService(requestUrl, jsonStr, username, uri, result);
                semaphore.release();// 释放信号灯
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void runService(String requestUrl, String jsonStr, String username, String uri, SynResult result) {
        String requestPath = null;
        try {
            jsonStr = jsonStr.substring(0, jsonStr.lastIndexOf("----"));
            JSONObject jsonObject = (JSONObject) JSONObject.parse(jsonStr);
            String originId = (String) jsonObject.get("originId");
            boolean hasKey = redisTemplate.hasKey(originId);
            if (!hasKey) {
//                redisTemplate.opsForValue().set(originId, originId);
                // 请求的参数列表
                Map<String, String> params = new HashMap<>();
                String now = String.valueOf(System.currentTimeMillis());
                String nonce = generateNonce(now);

                params.put(Constant.PARAM_APIKEY, apiKey);
                params.put(Constant.PARAM_TIMESTAMP, now);
                params.put(Constant.PARAM_NONCE, nonce);
                params.put(Constant.PARAM_SIGNATURE_METHOD, "HMACSHA256");

                logger.info(Constant.PARAM_APIKEY + ":" + apiKey);
                logger.info(Constant.PARAM_TIMESTAMP + ":" + now);
                logger.info(Constant.PARAM_NONCE + ":" + nonce);
                logger.info(Constant.PARAM_SIGNATURE_METHOD + ":" + "HMACSHA256");

                byte[] input = jsonStr.getBytes(java.nio.charset.Charset.forName("utf-8"));
                MessageDigest messageDigest = MessageDigest.getInstance("MD5");
                messageDigest.update(input);
                byte[] md5Bytes = messageDigest.digest();
                BASE64Encoder b64Encoder = new BASE64Encoder();
                String contentMD5 = b64Encoder.encode(md5Bytes);
                logger.info("contentMD5: " + contentMD5);
                String BuildRequestParameters = CommonUtil.getStandardRequestString(params);
                logger.info("BuildRequestParameters : " + BuildRequestParameters);
                String CanonicalizedQueryString = CommonUtil.percentEncode(BuildRequestParameters);
                logger.info("CanonicalizedQueryString : " + CanonicalizedQueryString);

                StringBuilder sb = new StringBuilder();
                sb.append("POST" + Constant.PARAM_CONNECTOR);
                sb.append(CanonicalizedQueryString);

                logger.info("HTTPMethod: POST");
                logger.info("SignStr: " + sb.toString());
                String sign = HmacShaUtil.calculateRFC2104HMAC(secretKey, sb.toString(), "HMACSHA256");
                logger.info("secretKey:Z1F3Y1lPbHFUOENuV3drMmJucGpwUT");
                logger.info("signature:" + sign);

                String Authorization = "Basic " + sign;

                HttpHeaders requestHeaders = new HttpHeaders();
                requestHeaders.add("Authorization", Authorization);
                if (!StringUtils.isEmpty(username)) {
                    requestHeaders.add("userName", username);
                } else {
                    requestHeaders.add("userName", "dax_master");
                }
                requestHeaders.setContentType(MediaType.APPLICATION_JSON);
                requestHeaders.add("Accept", MediaType.APPLICATION_JSON.toString());

                HttpEntity<String> requestEntity = new HttpEntity<>(jsonStr, requestHeaders);

//            String url = "http://192.168.0.154:18002/" + uri;
                String url;
                if (ObjectUtil.isEmpty(uri)) {
                    url = requestUrl;
                } else {
                    url = requestUrl + uri;
                }
                requestPath = url;
//            String url = requestUrl + uri;
                String param = "?" + Constant.PARAM_APIKEY + "=" + apiKey + "&" + Constant.PARAM_TIMESTAMP + "="
                        + now + "&" + Constant.PARAM_NONCE + "=" + nonce + "&" + Constant.PARAM_SIGNATURE_METHOD
                        + "=HMACSHA256";

                RestTemplate client = new RestTemplate();
                Date start = new Date();
                logger.info("请求开始时间:" + ":" + DateUtil.format(start));
                client.getMessageConverters().set(1, new StringHttpMessageConverter(StandardCharsets.UTF_8));
                ResponseEntity<String> response = client.postForEntity(url + param, requestEntity, String.class);
                Resp<?> resp = JsonMapper.nonDefaultMapper().fromJson(response.getBody(), Resp.class);
                String message = null;
                if (resp.getCode() == RespCode.OK && response.getStatusCodeValue() == RespCode.OK) {
                    result.getSuccessCount().getAndIncrement();
                    logger.info("请求成功耗时" + ":" + (System.currentTimeMillis() - start.getTime()) + "ms");
                } else {
                    if (resp.getResult() == null) {
                        message = resp.getMessage();
                    } else {
                        message = resp.getResult().toString();
                    }
                    result.getErrorCount().getAndIncrement();
                    logger.info("请求失败耗时" + ":" + (System.currentTimeMillis() - start.getTime()) + "ms");
                    sync_log.info(requestPath + "----" + jsonStr + "----" + StringUtils.stripToEmpty(username) + "--||--" + message);
                }
            } else {
                // 重复数据
                dump_log.info(jsonStr + "----" + username);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            sync_log.error(requestPath + "----" + jsonStr + "----" + StringUtils.stripToEmpty(username));
        }
    }
}
